package com.adee.java;

import com.adee.util.Utils;

import java.util.UUID;
import java.util.concurrent.*;

/**
 * https://www.cnblogs.com/hansc-blog/p/10645748.html
 *
 *
 * CompletableFuture是java.util.concurrent库在java 8中新增的主要工具，同传统的Future相比，其支持流式计算、函数式编程、
 * 完成通知、自定义异常处理等很多新的特性。
 * CompletableFuture 和 FutureTask 同属于 Future 接口的实现类，都可以获取线程的执行结果。
 *
 * FutureTask的局限性
 * FutureTask虽然提供了用来检查任务是否执行完成、等待任务执行结果、获取任务执行结果的方法，但是这些特色并不足以让我们写出简洁的
 * 并发代码，比如它并不能清楚地表达多个FutureTask之间的关系。另外，为了从Future获取结果，我们必须调用get()方法，而该方法还是会
 * 在任务执行完毕前阻塞调用线程，这明显不是我们想要的。 我们真正想要的是：
 * 可以将两个或者多个异步计算结合在一起变成一个，这包含两个或者多个异步计算是相互独立的情况，也包含第二个异步计算依赖第一个异步计算
 * 结果的情况。
 * 对反应式编程的支持，也就是当任务计算完成后能进行通知，并且可以以计算结果作为一个行为动作的参数进行下一步计算，而不是仅仅提供调用
 * 线程以阻塞的方式获取计算结果。 ·可以通过编程的方式手动设置（代码的方式）Future的结果；FutureTask不能实现让用户通过函数来设置
 * 其计算结果，而是在其任务内部来进行设置。
 *
 * 可以等多个Future对应的计算结果都出来后做一些事情。 为了克服FutureTask的局限性，以及满足我们对异步编程的需要，JDK8中提供了CompletableFuture。
 *
 *
 * 创建CompletableFuture
 * supplyAsync创建
 * CompletableFuture.supplyAsync()也可以用来创建CompletableFuture实例。通过该函数创建的CompletableFuture实例会异步执行
 * 当前传入的计算任务。在调用端，则可以通过get或join获取最终计算结果。
 * supplyAsync有两种签名：
 * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
 * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
 * 这两个方法的带 executor 的变种，表示让任务在指定的线程池中执行，不指定的话，通常任务是在 ForkJoinPool.commonPool() 线程池中执行的。
 *
 * runAsync创建
 * CompletableFuture.runAsync()也可以用来创建CompletableFuture实例。与supplyAsync()不同的是，runAsync()传入的任务要求
 * 是Runnable类型的，所以没有返回值。因此，runAsync适合创建不需要返回值的计算任务。同supplyAsync()类似，
 * runAsync()也有两种签名：
 * public static CompletableFuture<Void> runAsync(Runnable runnable)
 * public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
 *
 * 同Future相比，CompletableFuture最大的不同是支持流式（Stream）的计算处理，多个任务之间，可以前后相连，从而形成一个计算流。
 * 比如：任务1产生的结果，可以直接作为任务2的入参，参与任务2的计算，以此类推。
 * CompletableFuture中常用的流式连接函数包括：
 * thenApply——有入参有返回
 * thenApplyAsync——有入参有返回
 * thenAccept——有入参无返回
 * thenAcceptAsync——有入参无返回
 * thenRun——无入参无返回
 * thenRunAsync——无入参无返回
 * thenCombine
 * thenCombineAsync
 * thenCompose
 * thenComposeAsync
 * whenComplete
 * whenCompleteAsync
 * handle
 * handleAsync
 * 其中，带Async后缀的函数表示需要连接的后置任务会被单独提交到线程池中，从而相对前置任务来说是异步运行的。除此之外，两者没有其他区别。
 *
 * thenCombine / thenAcceptBoth / runAfterBoth互相不依赖
 * 这三个方法都是将两个CompletableFuture组合起来，只有这两个都正常执行完了才会执行某个任务，区别在于，thenCombine会
 * 将两个任务的执行结果作为方法入参传递到指定方法中，且该方法有返回值；thenAcceptBoth同样将两个任务的执行结果作为方法入参，
 * 但是无返回值；runAfterBoth没有入参，也没有返回值。注意两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。
 *
 */
public class Test022_2_CompletableFuture {
    static ThreadPoolExecutor tp = new ThreadPoolExecutor(10, 20,
            30, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(50),
            new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        test5();

    }

    // CompletableFuture.supplyAsync  有返回值
    public static void test1() throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
            Utils.sleep(5);
            return UUID.randomUUID().toString();
        }, tp);
        String s = cf.get();
        System.out.println(s);
    }

    // CompletableFuture.runAsync  无返回值
    public static void test2() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
            Utils.sleep(5);
        }, tp);
        System.out.println(cf.get());  // null
    }

    // CompletableFuture.runAsync  无返回值
    public static void test3() throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            Utils.sleep(5);
            return UUID.randomUUID().toString();
        }, tp);
        CompletableFuture<String> cf2 = cf1.thenApplyAsync((s) -> {
            return s + " +++cf2";
        });
        // System.out.println(cf2.get());  // 48c9f014-0d5a-4eeb-aaee-dd5e58d9e7c1 +++cf2
        // CompletableFuture.whenComplete 任务执行完回调
        cf2.whenComplete((r,e)->{
            if(e != null) {
                System.out.println("compute failed!");
            } else {
                System.out.println("result: " + r); // result: 140c3b99-1d1e-49b9-849b-d1aaa8a5421a +++cf2
            }
        });
    }

    // thenCombine
    public static void test4() throws ExecutionException, InterruptedException {
        long m = System.currentTimeMillis();
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            Utils.sleep(2);
            return 1+2;
        }, tp);
        CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
            Utils.sleep(3);
            return 3+4;
        }, tp);
        CompletableFuture<Integer> cf3 = cf1.thenCombine(cf2, (r1, r2) -> r1 + r2);
        Integer r = cf3.get();
        long t = System.currentTimeMillis() - m;
        System.out.println("计算结果：" + r + ", 耗时：" + t/1000 + "秒"); // 计算结果：10, 耗时：3秒
    }

    // 链式编程感知任务执行完成
    public static void test5() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> {
                    // return 1 + 2;
                    return 1/0;
                }, tp)
                // whenComplete感知任务执行完成，不能修改任务结果
                .whenComplete((r, e) -> {
                    if (e != null) {
                        System.out.println("compute failed!");
                    } else {
                        System.out.println("result: " + r);
                    }
                }).exceptionally(e -> { // exceptionally可以修改任务结果
                    System.out.println(e.getMessage());
                    if (e instanceof RuntimeException) {
                        return 0;
                    } else {
                        return -1;
                    }
                });
        System.out.println(cf.get());
    }
}
